package com.sangsang.kafka.consumer;

import com.alibaba.fastjson.JSONObject;
import com.alibaba.fastjson.TypeReference;
import com.baomidou.mybatisplus.core.toolkit.CollectionUtils;
import com.sangsang.ck.basemapper.CustomMapper;
import com.sangsang.kafka.domain.dto.PayloadDto;
import com.sangsang.kafka.util.CDCUtils;
import com.sangsang.kafka.util.DataParseUtils;
import com.sangsang.kafka.util.GenericityUtil;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;

import java.util.ArrayList;
import java.util.List;
import java.util.TimeZone;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

/**
 * T:对应同步表的实体类
 * M ：对应实体类的mapper
 *
 * @author liutangqi
 * @date 2023/12/25 15:50
 */
public abstract class AbstractCDCConsumerTemplate<T, M extends CustomMapper<T>> {
    private static final Logger log = LoggerFactory.getLogger(AbstractCDCConsumerTemplate.class);

    @Autowired
    private M mapper;


    /**
     * kafka消费者
     *
     * @author liutangqi
     * @date 2023/12/26 13:12
     * @Param [records]
     **/
    protected abstract void consumer(List<ConsumerRecord<String, String>> records);


    /**
     * 同步数据
     *
     * @author liutangqi
     * @date 2023/12/25 15:55
     * @Param [records]
     **/
    protected void processData(List<ConsumerRecord<String, String>> records) {
        Long startTime = System.currentTimeMillis();
        //1. 转换数据结构
        List<PayloadDto<T>> payloadDtos = parseRecord(records);
        //2.获取删除了的数据主键集合
        List deleteIds = DataParseUtils.<T>parseDeleteIds(payloadDtos);
        //3.获取修改后需要落库的数据
        List<T> entities = DataParseUtils.<T>parse(payloadDtos);
        //4.获取当前数据库中存在的值(这部分数据发生了变更，先删除旧数据，再插入)
        List existIds = getEntities(entities);
        //5.合并需要删除的主键集合
        existIds.addAll(deleteIds);
        //6.正式开始删除数据（主要耗时）
        deleteDbData(existIds);
        //7.批量插入这批数据
        batchInsertEntity(entities);
        log.info("【数据同步】{} 本批数量:{} 合并后数量：{} 耗时：{} ms", this.getClass().getSimpleName(), records.size(), entities.size(), (System.currentTimeMillis() - startTime));
    }

    /**
     * 获取当前实体类里面在数据库中存在的id
     *
     * @author liutangqi
     * @date 2023/12/29 10:56
     * @Param [entities]
     **/
    private List<T> getEntities(List<T> entities) {
        List ids = entities.stream()
                .map(CDCUtils::getPrimaryKeyValue)
                .filter(f -> f != null)
                .collect(Collectors.toList());
        if (CollectionUtils.isEmpty(ids)) {
            return new ArrayList<>();
        }

        return mapper.selectExistIdByIds(ids);
    }


    /**
     * 删除ck库中需要删除的数据
     *
     * @author liutangqi
     * @date 2023/12/28 16:22
     * @Param [existIds 主键id集合]
     **/
    private void deleteDbData(List ids) {
        //自旋次数，用于日志输出
        int spinCount = 0;

        //没有需要删除的数据则不处理
        if (CollectionUtils.isEmpty(ids)) {
            return;
        }

        //6.1异步删除
        mapper.ckBatchDeleteByIds(ids);
        //6.2 自旋等待 删除完毕
        while (mapper.ckQueryMutationsCount() > 0) {
            try {
                spinCount++;
                TimeUnit.MILLISECONDS.sleep(100);
                if (spinCount > 10) {
                    log.warn("【数据同步】<> mutations自旋次数：{}", this.getClass().getSimpleName(), spinCount);
                }
            } catch (InterruptedException e) {
                log.error("【数据同步】异常", e);
                e.printStackTrace();
            }
        }
    }

    /**
     * 批量插入数据到ck库
     *
     * @author liutangqi
     * @date 2023/12/28 16:24
     * @Param [datas]
     **/
    private void batchInsertEntity(List<T> datas) {
        if (CollectionUtils.isNotEmpty(datas)) {
            mapper.batchInsert(datas);
        }
    }

    /**
     * 将kafka的消息反序列化成对应实体类
     *
     * @author liutangqi
     * @date 2023/12/25 15:53
     * @Param [records]
     **/
    private List<PayloadDto<T>> parseRecord(List<ConsumerRecord<String, String>> records) {
        //转换时区
        TimeZone.setDefault(TimeZone.getTimeZone("UTC"));
        //获取当前泛型实体类
        Class<?> tCls = GenericityUtil.getClassT(this, 0);
        return records.stream()
                .filter(f -> f.value() != null)
                .map(ConsumerRecord::value)
                .map(m -> JSONObject.parseObject(m, new TypeReference<PayloadDto<T>>(tCls) {
                })).collect(Collectors.toList());
    }
}
